#include "transContentStreamMemoryOut.hpp"

////////////////////////////////////////////////////////////////////////////////////////////////////////////
/// send user supplied data to remote port
///		flag must be 0 or 'SEND_FLAG_NONBLICK', else will be reset to 0
///		reset all parameter to instance value, then notice the send thread to start
///		after send, call callback or return value
///
///	parameters:
///		flag: send flag, set to block or not
///		buffer: the send out data buffer
///		bufferLen: the buffer length
///		timeout: the send used time
///		sscb: the send finihsed callback
///
///	return :
///		E_TRANC_ParamError: given parameter error
///		E_TRANC_NONBLOCK: no enough buffer avaliable, and set no block, so return
///		
int TransStreamOutMemory::send(int flag, char* buffer, int bufferLen, int timeout, sendSyncCallback sscb){
	timespec ts;
	if(buffer == NULL || bufferLen <= 0){
		logd_error("stream out memory, supplied buffer or length error");
		return E_TRANC_ParamError;
	}
	if(sending == 1){
		logd_error("steram out memroy, there is still a sending tack running");
		return E_TRANC_STATE_BUSY;
	}
	sending = 1;
	/// check left bytes can hold the bufflen
	if(bufferLen > getLeftSpaceBytes() && flag == SEND_FLAG_NONBLICK){
		logd_error("stream out memory, send data length larger than the buffer size, and the flag set to nonblock");
		return E_TRANC_NONBLOCK;
	}
	this->timeout = timeout;
	this->sscb = sscb;

	/// set data to send 
	mOutData = buffer;
	mOutDataLen = bufferLen;

	startToSend();

	if(this->sscb != NULL){
		logd_info("stream out memroy, send data use async style, the caller will not be blocked");
		return E_TRANC_NONBLOCK;
	}

	ts = TransContentAssis::TimeTp2Ts(this->timeout);
	/// wait
	pthread_mutex_lock(&waitMutex);
	if(pthread_cond_timedwait(&waitCond, &waitMutex, &ts) == ETIMEDOUT){
		logd_error("stream out memroy, send timeout, stop the send thread");
		sending = 0;
		return E_TRANC_TIMEOUT;
	}
	pthread_mutex_unlock(&waitMutex);

	logd_info("stream out memory, send finihsed");
	return mOutDataLen;
}

void startToSend(){
	pthread_mutex_lock(&dataAvalaMutex);
	pthread_cond_signal(&dataAvalaCond);
	pthread_mutex_unlock(&dataAvalaMutex);
}

void sendworker(){
	while(working){
		pthread_mutex_lock(&dataAvalaMutex);
		pthread_cond_wait(&dataAvalaCond, &dataAvalaMutex);
		pthread_mutex_unlock(&dataAvalaMutex);
		sendDataOut();
	}
}

int sendDataOut(){
	/// init the send bytes to data length
	int __leftBytes = mOutDataLen;
	/// get a free buffer index
	int __bufferIndex = -1;

	char* __buffer = NULL;
	/// already send bytes
	int __sendBytes = 0;

	int __bufferSize = TransContentSetting::getMemoryBufferSize()-50;
	
	debug_enter();
	
	while(__leftBytes > 0){
		/// get avaliable buffer
		logd_info("stream out, send data out, wait to get the left count > 0");
		while(mLeftCount == 0 && sending == 1){
			pthread_mutex_lock(&waitAvaliableBufferMutex);
			if(pthread_cond_timedwait(&waitAvaliableBufferCond, &waitAvaliableBufferMutex, &waitAvalbaileBufferTS) == ETIMEDOUT){
				logd_error("stream out, send data out, wait avaliable buffer timeout, set test remote port flag");
				checkRemoteHeartBeat = 1;
				continue;
			}
			pthread_mutex_unlock(&waitAvaliableBufferMutex);
		}

		/// check send data cancled
		if(sending == 0){
			logd_warmming("stream out, send data out, sending canceled by timedout!");
			break;
		}

		/// get a buffer from memory pool
		while(1){
		__bufferIndex = TransContentMemory::getInstance()->getBuffer(&__buffer);
		if(__bufferIndex < 0){
			logd_warmming("stream out, send data out, get buffer from memory pool, no avaliable buffer, usleep");
			usleep(TIME_TO_WAIT_AVALIABLE_BUFFER);
			continue;
		}
		logd_info("stream out, send data out, get buffer sucessfully, buffer index=%d", __bufferIndex);
		break;
		}

		/// fill data
		if(__bufferSize > __leftBytes){
			*(short*(buffer+24)) = TransContentAssis::INTL2N(__leftBytes);
			memcpy(buffer+26, mOutData + __sendBytes, __leftBytes);
			__sendBytes = __leftBytes;
			__leftBytes = 0;
		}else{
			*(short*(buffer+24)) = TransContentAssis::INTL2N(__bufferSize);
			memcpy(buffer+26, mOutData + __sendBytes, __bufferSize);
			__sendBytes = __bufferSize;
			__leftBytes -= __sendBytes;
		}

		/// construct packet information
		*(short*(buffer)) = __sendBytes + 24; //total len
		buffer[2] = (char)message_type_out_data_stream; //main code
		*(NodeAddr*(buffer+3)) = target; //target
		buffer[11] = CLIENT_STACK_FLAG_BITS_FIXEDPATH; //flag
		*(short*(buffer+12)) = TransContentAssis::SHORTL2N(__sendBytes + 12); //data record len
		buffer[14] = (char)message_type_out_data_stream; //main code
		buffer[15] = (char)message_type_data_stream_sub_dataIn; //sub code
		*(int*(buffer+16)) = TransContentAssis::INTL2N(remotePort); //remote port
		*(int*(buffer+20)) = TransContentAssis::INTL2N(mOutStreamValue++); //stream id

		/// add packet to pool
		mRecordBuffer[mInsertIndex] = __bufferIndex;
		--mLeftCount;
		/// send packet out
		sendPacketOut();
		mInsertIndex = (mInsertIndex+1) % mCapacity;
	}
	sending = 0;
	sendFinished();
	return E_TRANC_SUCESS;
}

////////////////////////////////////////////////////////////////////////
/// send a packet to remote
///////////////////////////////////////////////////////////////////////
void sendPacketOut(){
	char* buffer;
	int length;
	/// reset the should send count
	++shouldSendCount;
	
	///send all should record when the alowdSendCount allowd
	while(allowSendCount > 0 && shouldSendCount>0){
		--allowSendCount;
		--shouldSendCount;
		buffer = TransContentMemory::getInstance()->indexToBuffer(startSendIndex);
		length = *(short*(buffer)) + 2;
		TransContentDispatcher::getInstance()->sendDataDataToStack(buffer, length);
		startSendIndex = (startSendIndex + 1) % mCapacity;
	}
}

int response(char* buffer, int index){
	int mainCode = buffer[0];
	int subCode = buffer[1];
	if(mainCode != message_type_out_data_stream){
		logd_error("stream port, rsponse, not support the main code");
		return RESPONSE_RESULT_POST_CONITNUE;
	}
	switch(subCode){
		case message_type_data_stream_sub_dataACK:
			logd_info("stream port, response, sub message data ack");
			ackResponse(buffer);
			return RESPONSE_RESULT_RELEASEBUFFER;
		default:
			logd_error("stream port, response, not support sub code: %d", subCode);
			return RESPONSE_RESULT_POST_CONITNUE;
	}
	logd_error("stream port, response, should not reached here!!");
	return RESPONSE_RESULT_NOACTION;
}

int ackResponse(char* buffer){
	int streamId = TransContentAssis::INTN2L(*(int*(buffer+6)));
	int count = streamId - mAckStartFragmentId;
	while(count >= 0){
		mRecordBuffer[mAckStartIndex] = -1;
		++mLeftCount;
		++mAckStartFragmentId;
		++mAckStartIndex;
		++allowSendCount;
	}
	pthread_mutex_lock(&waitAvaliableBufferMutex);
	pthread_cond_signal(&waitAvaliableBufferCond);
	pthread_mutex_unlock(&waitAvaliableBufferMutex);
	return E_TRANC_SUCESS;
}

void ReSendFirstRecord(){
	char* buffer = TransContentMemory::getInstance()->indexToBuffer(mAckStartIndex);
	int length = *(short*(buffer)) + 2;
	logd_info("stream port, resend packet, send out bytes: length=%d", length);
	TransContentDispatcher::getInstance()->sendDataDataToStack(buffer, length);
}

void sendFinished(){
	pthread_t pid;
	if(this->sscb != NULL){
		pthread_create(&pid, NULL, finishRoute, this);
		pthread_detach(pid);
		logd_info("stream port, send out finished, async");
		return;
	}
	logd_info("stream port, send out finihed, sync");
	pthread_mutex_lock(&waitMutex);
	pthread_cond_signal(&waitCond);
	pthread_mutex_unlock(&waitMutex);
}

void finishCallback(){
	this->sscb(1);
}
